"""
数据源管理服务
负责数据源的配置、监控和管理
"""
import pandas as pd
from typing import Dict, Any, List, Optional, Union
from datetime import datetime
from sqlalchemy.orm import Session
from sqlalchemy.sql import func

from models.data_source import DataSource, DataSourceSyncLog, DataSourcePreviewLog, DatabaseConfig, ExcelConfig, DataSourceType
from models.analysis_task import AnalysisTask
from services.data_source_connection import DataSourceConnection
from services.excel_data_processor import excel_processor
from core.logging_config import get_logger
from core.exceptions import DatabaseError, ValidationError

logger = get_logger("data_source_manager")


class DataSourceManager:
    """数据源管理器"""
    
    def __init__(self):
        self.connection_cache: Dict[int, DataSourceConnection] = {}
        self.last_health_check = {}
    
    def create_data_source(self, db: Session, user_id: int, data_source_config: Dict[str, Any]) -> DataSource:
        """创建数据源"""
        try:
            # 验证配置
            self._validate_data_source_config(data_source_config)
            
            # 创建数据源记录
            data_source = DataSource(
                name=data_source_config["name"],
                description=data_source_config.get("description", ""),
                source_type=data_source_config["type"],
                creator_id=user_id,
                status="inactive"
            )
            
            db.add(data_source)
            db.commit()
            db.refresh(data_source)
            
            # 根据类型创建配置
            if data_source_config["type"] == "database":
                db_config = DatabaseConfig.from_config(data_source.id, data_source_config["config"])
                db.add(db_config)
            elif data_source_config["type"] == "excel":
                excel_config = ExcelConfig(
                    data_source_id=data_source.id,
                    **data_source_config["config"]
                )
                db.add(excel_config)
            
            db.commit()
            
            # 测试连接
            test_result = self.test_data_source_connection(data_source)
            
            # 更新状态
            if test_result["success"]:
                data_source.status = "active"
                data_source.last_sync_result = "连接测试成功"
                data_source.last_sync_time = func.now()
            
            db.commit()
            
            logger.info(f"数据源创建成功: {data_source.name} (ID: {data_source.id})")
            return data_source
            
        except Exception as e:
            db.rollback()
            logger.error(f"数据源创建失败: {str(e)}")
            raise DatabaseError(f"数据源创建失败: {str(e)}")
    
    def update_data_source(self, db: Session, data_source_id: int, update_config: Dict[str, Any]) -> DataSource:
        """更新数据源"""
        try:
            data_source = db.query(DataSource).filter(DataSource.id == data_source_id).first()
            if not data_source:
                raise ValidationError(f"数据源不存在: {data_source_id}")
            
            # 更新基本信息
            if "name" in update_config:
                data_source.name = update_config["name"]
            if "description" in update_config:
                data_source.description = update_config["description"]
            
            # 如果配置有变化，需要重新测试连接
            if "config" in update_config:
                # 更新配置
                if data_source.source_type == "database":
                    db_config = data_source.database_config
                    
                    if db_config:
                        # 更新现有配置
                        config = update_config["config"].copy()
                        if "database_type" in config:
                            config["db_type"] = config.pop("database_type")
                        if "type" in config:
                            config["db_type"] = config.pop("type")
                        
                        for key, value in config.items():
                            setattr(db_config, key, value)
                    else:
                        # 创建新配置
                        db_config = DatabaseConfig.from_config(data_source.id, update_config["config"])
                        db.add(db_config)
                elif data_source.source_type == "excel":
                    excel_config = data_source.excel_config
                    if excel_config:
                        for key, value in update_config["config"].items():
                            setattr(excel_config, key, value)
                    else:
                        excel_config = ExcelConfig(
                            data_source_id=data_source.id,
                            **update_config["config"]
                        )
                        db.add(excel_config)
                
                data_source.update_status("inactive")
                
                # 清除缓存连接
                if data_source_id in self.connection_cache:
                    self.connection_cache[data_source_id].close()
                    del self.connection_cache[data_source_id]
            
            db.commit()
            
            # 重新测试连接
            if "config" in update_config:
                test_result = self.test_data_source_connection(data_source)
                if test_result["success"]:
                    data_source.update_status("active", "连接测试成功")
                
                db.commit()
            
            logger.info(f"数据源更新成功: {data_source.name}")
            return data_source
            
        except Exception as e:
            db.rollback()
            logger.error(f"数据源更新失败: {str(e)}")
            raise DatabaseError(f"数据源更新失败: {str(e)}")
    
    def delete_data_source(self, db: Session, data_source_id: int) -> bool:
        """删除数据源"""
        try:
            data_source = db.query(DataSource).filter(DataSource.id == data_source_id).first()
            if not data_source:
                raise ValidationError(f"数据源不存在: {data_source_id}")
            
            # 清除缓存连接
            if data_source_id in self.connection_cache:
                self.connection_cache[data_source_id].close()
                del self.connection_cache[data_source_id]
            
            # 由于SQLite外键约束限制，需要按照以下顺序删除：
            # 1. 先删除分析任务
            db.query(AnalysisTask).filter(
                AnalysisTask.data_source_id == data_source_id
            ).delete()
            
            # 2. 删除预览日志记录
            db.query(DataSourcePreviewLog).filter(
                DataSourcePreviewLog.data_source_id == data_source_id
            ).delete()
            
            # 3. 删除连接历史记录
            db.query(DataSourceSyncLog).filter(
                DataSourceSyncLog.data_source_id == data_source_id
            ).delete()
            
            # 4. 删除配置信息
            if data_source.source_type == DataSourceType.DATABASE:
                db.query(DatabaseConfig).filter(
                    DatabaseConfig.data_source_id == data_source_id
                ).delete()
            elif data_source.source_type == DataSourceType.EXCEL:
                db.query(ExcelConfig).filter(
                    ExcelConfig.data_source_id == data_source_id
                ).delete()
            
            # 5. 删除数据源
            db.delete(data_source)
            db.commit()
            
            logger.info(f"数据源删除成功: {data_source.name}")
            return True
            
        except Exception as e:
            db.rollback()
            logger.error(f"数据源删除失败: {str(e)}")
            raise DatabaseError(f"数据源删除失败: {str(e)}")
    
    def get_data_source(self, db: Session, data_source_id: int) -> Optional[DataSource]:
        """获取数据源"""
        return db.query(DataSource).filter(DataSource.id == data_source_id).first()
    
    def get_data(self, db: Session, data_source_id: int) -> Dict[str, Any]:
        """获取数据源中的数据
        
        Args:
            db: 数据库会话
            data_source_id: 数据源ID
            
        Returns:
            Dict[str, Any]: 包含数据的字典
        """
        try:
            # 获取数据源
            data_source = self.get_data_source(db, data_source_id)
            if not data_source:
                raise ValidationError(f"数据源不存在: {data_source_id}")
            
            # 检查数据源类型
            if data_source.source_type != DataSourceType.DATABASE:
                raise ValidationError(f"只支持数据库类型的数据源，当前类型: {data_source.source_type}")
            
            # 获取数据库配置
            if not data_source.database_config:
                raise ValidationError("数据库配置不存在")
            
            db_config = data_source.database_config
            
            # 检查表名
            if not db_config.table_name:
                raise ValidationError("数据库表名未配置")
            
            logger.info(f"开始获取数据源 {data_source_id} 的数据，表名: {db_config.table_name}")
            
            # 创建数据库连接
            connection = DataSourceConnection(db_config.to_dict())
            
            # 测试连接
            test_result = connection.test_connection()
            if not test_result["success"]:
                raise DatabaseError(f"数据库连接失败: {test_result['message']}")
            
            # 获取表数据
            result = connection.get_table_data(db_config.table_name)
            
            logger.info(f"成功获取数据源 {data_source_id} 的数据，记录数: {result.get('row_count', 0)}")
            
            return {
                "success": True,
                "data": result.get("data", []),
                "record_count": result.get("record_count", 0),
                "column_count": result.get("column_count", 0),
                "table_name": db_config.table_name,
                "message": "数据获取成功"
            }
            
        except ValidationError as e:
            logger.error(f"数据验证失败: {str(e)}")
            raise
        except DatabaseError as e:
            logger.error(f"数据库操作失败: {str(e)}")
            raise
        except Exception as e:
            logger.error(f"获取数据失败: {str(e)}")
            raise DatabaseError(f"获取数据失败: {str(e)}")
    
    def get_data_sources(self, db: Session, user_id: Optional[int] = None, 
                         source_type: Optional[str] = None, 
                         status: Optional[str] = None) -> List[DataSource]:
        """获取数据源列表"""
        query = db.query(DataSource)
        
        if user_id:
            query = query.filter(DataSource.creator_id == user_id)
        
        if source_type:
            query = query.filter(DataSource.source_type == source_type)
        
        if status:
            query = query.filter(DataSource.status == status)
        
        return query.order_by(DataSource.created_at.desc()).all()
    
    def test_data_source_connection(self, data_source: DataSource) -> Dict[str, Any]:
        """测试数据源连接"""
        try:
            if data_source.source_type == "database":
                # 创建数据库连接
                config = data_source.get_config()
                if not config:
                    return {"success": False, "message": "数据库配置不存在"}
                # 创建数据库连接
                connection = DataSourceConnection(config)
                test_result = connection.test_connection()
                
                # 缓存连接
                if test_result["success"]:
                    self.connection_cache[data_source.id] = connection
                
                return test_result
                
            elif data_source.source_type == "excel":
                # Excel文件验证
                config = data_source.get_config()
                file_path = config.get("file_path")
                if not file_path:
                    return {"success": False, "message": "Excel文件路径不能为空"}
                
                validation_result = excel_processor.validate_file(file_path)
                
                if validation_result["valid"]:
                    # 尝试读取文件
                    read_result = excel_processor.read_excel_file(file_path)
                    if read_result["success"]:
                        return {
                            "success": True,
                            "message": "Excel文件读取成功",
                            "file_info": validation_result,
                            "data_info": read_result["data"]
                        }
                    else:
                        return {"success": False, "message": read_result["message"]}
                else:
                    return {"success": False, "message": validation_result["message"]}
            
            else:
                return {"success": False, "message": f"不支持的数据源类型: {data_source.source_type}"}
                
        except Exception as e:
            logger.error(f"测试数据源连接失败: {str(e)}")
            return {"success": False, "message": f"连接测试失败: {str(e)}"}
    
    def get_data_source_connection(self, data_source: DataSource) -> Optional[DataSourceConnection]:
        """获取数据源连接"""
        try:
            # 检查缓存
            if data_source.id in self.connection_cache:
                connection = self.connection_cache[data_source.id]
                if connection.is_connected:
                    return connection
            
            # 创建新连接
            if data_source.source_type == "database":
                config = data_source.get_config()
                if not config:
                    raise DatabaseError("数据库配置不存在")
                
                connection = DataSourceConnection(config)
                test_result = connection.test_connection()
                
                if test_result["success"]:
                    self.connection_cache[data_source.id] = connection
                    return connection
                else:
                    raise DatabaseError(f"连接测试失败: {test_result['message']}")
            
            elif data_source.source_type == "excel":
                # Excel数据源不需要连接，直接返回None
                return None
            
            else:
                raise DatabaseError(f"不支持的数据源类型: {data_source.source_type}")

        except Exception as e:
            logger.error(f"获取数据源连接失败: {str(e)}")
            raise DatabaseError(f"获取连接失败: {str(e)}")
    
    def get_data_sources_statistics(self, db: Session, user_id: Optional[int] = None) -> Dict[str, Any]:
        """获取数据源统计信息"""
        try:
            # 构建查询条件
            query = db.query(DataSource)
            if user_id:
                query = query.filter(DataSource.creator_id == user_id)
            
            data_sources = query.all()
            
            # 统计数据
            statistics = {
                "total_count": len(data_sources),
                "active_count": 0,
                "inactive_count": 0,
                "error_count": 0,
                "database_count": 0,
                "excel_count": 0,
                "by_type": {},
                "by_status": {},
                "recent_created": [],
                "recent_connected": []
            }
            
            # 按类型和状态统计
            for ds in data_sources:
                # 按类型统计
                source_type = ds.source_type or "unknown"
                statistics["by_type"][source_type] = statistics["by_type"].get(source_type, 0) + 1
                
                # 按状态统计
                status = ds.status or "unknown"
                statistics["by_status"][status] = statistics["by_status"].get(status, 0) + 1
                
                # 状态计数
                if ds.status == "active":
                    statistics["active_count"] += 1
                elif ds.status == "inactive":
                    statistics["inactive_count"] += 1
                elif ds.status == "error":
                    statistics["error_count"] += 1
                
                # 按数据源类型计数
                if ds.source_type == "database":
                    statistics["database_count"] += 1
                elif ds.source_type == "excel":
                    statistics["excel_count"] += 1
                
                # 最近创建的数据源
                if ds.created_at:
                    statistics["recent_created"].append({
                        "id": ds.id,
                        "name": ds.name,
                        "created_at": ds.created_at.isoformat(),
                        "type": ds.source_type
                    })
                
                # 最近连接的数据源
                if ds.last_sync_time:
                    statistics["recent_connected"].append({
                        "id": ds.id,
                        "name": ds.name,
                        "last_connected": ds.last_sync_time.isoformat(),
                        "status": ds.status
                    })
            
            # 排序并限制数量
            statistics["recent_created"].sort(
                key=lambda x: x["created_at"], 
                reverse=True
            )
            statistics["recent_created"] = statistics["recent_created"][:5]
            
            statistics["recent_connected"].sort(
                key=lambda x: x["last_connected"], 
                reverse=True
            )
            statistics["recent_connected"] = statistics["recent_connected"][:5]
            
            # 计算连接率
            if statistics["total_count"] > 0:
                statistics["connection_rate"] = round(
                    (statistics["active_count"] / statistics["total_count"]) * 100, 1
                )
            else:
                statistics["connection_rate"] = 0
            
            logger.info(f"数据源统计信息: {statistics}")
            return statistics
            
        except Exception as e:
            logger.error(f"获取数据源统计信息失败: {str(e)}")
            raise DatabaseError(f"获取统计信息失败: {str(e)}")
    
    def get_data_source_stats(self, db: Session, data_source_id: int) -> Dict[str, Any]:
        """获取数据源统计信息"""
        try:
            data_source = self.get_data_source(db, data_source_id)
            if not data_source:
                raise ValidationError(f"数据源不存在: {data_source_id}")
            
            stats = {
                "data_source_id": data_source_id,
                "name": data_source.name,
                "type": data_source.source_type,
                "status": data_source.status,
                "created_at": data_source.created_at.isoformat(),
                "last_connected": data_source.last_sync_time.isoformat() if data_source.last_sync_time else None,
                "usage_count": 0,
                "connection_history": []
            }
            
            # 获取连接历史
            history = db.query(DataSourceSyncLog).filter(
                DataSourceSyncLog.data_source_id == data_source_id
            ).order_by(DataSourceSyncLog.created_at.desc()).limit(10).all()
            
            stats["connection_history"] = [
                {
                    "id": h.id,
                    "status": h.status,
                    "message": h.error_message,
                    "created_at": h.created_at.isoformat()
                } for h in history
            ]
            
            # 根据数据源类型获取特定统计信息
            if data_source.source_type == "database":
                connection = self.get_data_source_connection(data_source)
                if connection:
                    try:
                        tables = connection.get_table_list()
                        stats["database_info"] = {
                            "table_count": len(tables),
                            "tables": tables[:10]  # 只返回前10个表
                        }
                    except Exception as e:
                        stats["database_info"] = {"error": str(e)}
            
            elif data_source.source_type == "excel":
                config = data_source.get_config()
                file_path = config.get("file_path")
                if file_path:
                    try:
                        read_result = excel_processor.read_excel_file(file_path)
                        if read_result["success"]:
                            # 统计所有工作表的数据
                            total_rows = 0
                            total_columns = 0
                            for sheet_data in read_result["data"].values():
                                total_rows += sheet_data["row_count"]
                                total_columns += sheet_data["column_count"]
                            
                            stats["excel_info"] = {
                                "total_rows": total_rows,
                                "total_columns": total_columns,
                                "sheet_count": len(read_result["data"]),
                                "sheets": list(read_result["data"].keys())
                            }
                    except Exception as e:
                        stats["excel_info"] = {"error": str(e)}
            
            return stats
            
        except Exception as e:
            logger.error(f"获取数据源统计信息失败: {str(e)}")
            raise DatabaseError(f"获取统计信息失败: {str(e)}")
    
    def record_connection_history(self, db: Session, data_source_id: int, 
                                 status: str, message: str, details: Dict[str, Any] = None):
        """记录连接历史"""
        try:
            history = DataSourceSyncLog(
                data_source_id=data_source_id,
                sync_type="connection_test",
                status=status,
                start_time=datetime.utcnow(),
                error_message=message,
                error_details=details or {}
            )
            
            db.add(history)
            db.commit()
            
        except Exception as e:
            logger.error(f"记录连接历史失败: {str(e)}")
    
    def health_check_all_data_sources(self, db: Session) -> Dict[str, Any]:
        """健康检查所有数据源"""
        try:
            data_sources = db.query(DataSource).filter(DataSource.status == "active").all()
            
            results = {
                "total_count": len(data_sources),
                "healthy_count": 0,
                "unhealthy_count": 0,
                "details": []
            }
            
            for data_source in data_sources:
                try:
                    test_result = self.test_data_source_connection(data_source)
                    
                    # 更新数据源状态
                    if test_result["success"]:
                        data_source.update_status("active")
                        results["healthy_count"] += 1
                    else:
                        data_source.update_status("inactive")
                        results["unhealthy_count"] += 1
                    
                    # 记录历史
                    self.record_connection_history(
                        db, 
                        data_source.id, 
                        "success" if test_result["success"] else "failed",
                        test_result.get("message", ""),
                        test_result
                    )
                    
                    results["details"].append({
                        "id": data_source.id,
                        "name": data_source.name,
                        "type": data_source.source_type,
                        "status": "active" if test_result["success"] else "inactive",
                        "message": test_result.get("message", ""),
                        "last_checked": datetime.utcnow().isoformat()
                    })
                    
                except Exception as e:
                    data_source.status = "error"
                    results["unhealthy_count"] += 1
                    
                    results["details"].append({
                        "id": data_source.id,
                        "name": data_source.name,
                        "type": data_source.source_type,
                        "status": "error",
                        "message": str(e),
                        "last_checked": datetime.utcnow().isoformat()
                    })
            
            db.commit()
            
            logger.info(f"数据源健康检查完成: {results['healthy_count']}/{results['total_count']} 正常")
            return results
            
        except Exception as e:
            logger.error(f"数据源健康检查失败: {str(e)}")
            raise DatabaseError(f"健康检查失败: {str(e)}")
    
    def _validate_data_source_config(self, config: Dict[str, Any]):
        """验证数据源配置"""
        required_fields = ["name", "type"]
        
        for field in required_fields:
            if field not in config:
                raise ValidationError(f"缺少必需字段: {field}")
        
        # 验证特定类型的配置
        if config["type"] == "database":
            # 检查数据库类型字段（支持 database_type 或 type）
            db_config = config.get("config", {})
            if "database_type" not in db_config and "type" not in db_config:
                raise ValidationError("数据库数据源缺少必需字段: database_type 或 type")
            
            # 根据数据库类型验证必需字段
            db_type = db_config.get("database_type") or db_config.get("type")
            if db_type == "sqlite":
                if "database" not in db_config:
                    raise ValidationError("SQLite数据源缺少必需字段: database")
            else:
                # 其他数据库类型需要更多字段
                required_db_fields = ["host", "database"]
                for field in required_db_fields:
                    if field not in db_config:
                        raise ValidationError(f"数据库数据源缺少必需字段: {field}")
        
        elif config["type"] == "excel":
            if "file_path" not in config.get("config", {}):
                raise ValidationError("Excel数据源缺少文件路径")
    
    def cleanup_old_connections(self):
        """清理旧的连接"""
        try:
            current_time = datetime.utcnow()
            expired_connections = []
            
            for data_source_id, connection in self.connection_cache.items():
                # 如果连接超过1小时未使用，关闭它
                if hasattr(connection, 'last_used'):
                    if (current_time - connection.last_used).total_seconds() > 3600:
                        expired_connections.append(data_source_id)
                else:
                    # 如果没有last_used属性，直接关闭
                    expired_connections.append(data_source_id)
            
            for data_source_id in expired_connections:
                self.connection_cache[data_source_id].close()
                del self.connection_cache[data_source_id]
                logger.info(f"清理过期连接: 数据源ID {data_source_id}")
            
        except Exception as e:
            logger.error(f"清理连接失败: {str(e)}")
    
    def get_supported_data_source_types(self) -> List[Dict[str, Any]]:
        """获取支持的数据源类型"""
        return [
            {
                "type": "database",
                "name": "数据库",
                "description": "支持MySQL、PostgreSQL、SQLite等数据库",
                "config_template": {
                    "type": "mysql",
                    "host": "localhost",
                    "port": 3306,
                    "database": "",
                    "username": "",
                    "password": "",
                    "extra_params": {}
                }
            },
            {
                "type": "excel",
                "name": "Excel文件",
                "description": "支持Excel和CSV文件数据导入",
                "config_template": {
                    "file_path": "",
                    "sheet_name": "",
                    "has_header": True
                }
            }
        ]

    def get_data_source_schema(self, db: Session, data_source_id: int) -> List[Dict[str, Any]]:
        """获取数据源结构信息"""
        try:
            data_source = self.get_data_source(db, data_source_id)
            if not data_source:
                raise ValidationError(f"数据源不存在: {data_source_id}")
            
            schema_info = []
            
            if data_source.source_type == "database":
                # 获取数据库表结构
                connection = self.get_data_source_connection(data_source)
                if connection:
                    tables = connection.get_table_list()
                    for table in tables:
                        # table 是表信息字典，需要提取表名
                        actual_table_name = table['name'] if isinstance(table, dict) else table
                        table_schema = connection.get_table_schema(actual_table_name)
                        schema_info.append({
                            "table_name": actual_table_name,
                            "columns": table_schema.get("columns", []),
                            "record_count": table_schema.get("record_count", 0),
                            "column_count": table_schema.get("column_count", 0)
                        })
                        
            elif data_source.source_type == "excel":
                # 获取Excel文件结构
                config = data_source.get_config()
                if config:
                    file_path = config.get("file_path")
                    if file_path:
                        read_result = excel_processor.read_excel_file(file_path)
                        if read_result["success"]:
                            for sheet_name, sheet_data in read_result["data"].items():
                                schema_info.append({
                                    "table_name": sheet_name,
                                    "columns": sheet_data.get("columns", []),
                                    "record_count": sheet_data.get("row_count", 0),
                                    "column_count": sheet_data.get("column_count", 0)
                                })
            
            # 记录预览日志
            self._record_preview_log(
                db, 
                data_source_id, 
                "table_schema", 
                "success", 
                len(schema_info)
            )
            
            return schema_info
            
        except Exception as e:
            logger.error(f"获取数据源结构信息失败: {str(e)}")
            
            # 记录错误日志
            if 'data_source_id' in locals():
                self._record_preview_log(
                    db, 
                    data_source_id, 
                    "table_schema", 
                    "failed", 
                    0, 
                    error_message=str(e)
                )
            
            raise DatabaseError(f"获取数据源结构信息失败: {str(e)}")
    
    def _clean_numpy_types(self, data):
        """清理numpy类型，转换为Python原生类型以便JSON序列化"""
        import numpy as np
        
        if isinstance(data, dict):
            return {key: self._clean_numpy_types(value) for key, value in data.items()}
        elif isinstance(data, list):
            return [self._clean_numpy_types(item) for item in data]
        elif isinstance(data, np.integer):
            return int(data)
        elif isinstance(data, np.floating):
            return float(data)
        elif isinstance(data, np.bool_):
            return bool(data)
        elif isinstance(data, np.ndarray):
            return data.tolist()
        else:
            return data
    
    def preview_data_source(self, db: Session, data_source_id: int, 
                           table_name: Optional[str] = None, 
                           sheet_name: Optional[str] = None,
                           limit: int = 100, 
                           offset: int = 0) -> Dict[str, Any]:
        """预览数据源数据"""
        try:
            import time
            start_time = time.time()
            
            data_source = self.get_data_source(db, data_source_id)
            if not data_source:
                raise ValidationError(f"数据源不存在: {data_source_id}")
            
            preview_data = {
                "table_name": table_name or sheet_name or "data",
                "columns": [],
                "data": [],
                "total_records": 0,
                "returned_records": 0,
                "offset": offset,
                "limit": limit,
                "execution_time": 0
            }
            
            if data_source.source_type == "database":
                # 预览数据库数据
                connection = self.get_data_source_connection(data_source)
                if connection:
                    logger.info(f"数据库连接成功，开始预览数据")
                    
                    if not table_name:
                        # 获取第一个表
                        tables = connection.get_table_list()
                        logger.info(f"数据库表列表: {tables}")
                        
                        if tables:
                            # tables 是表信息字典列表，取第一个表的 'name' 字段
                            table_name = tables[0]['name'] if isinstance(tables[0], dict) else tables[0]
                            logger.info(f"自动选择表名: {table_name}")
                        else:
                            logger.warning("数据库中没有找到任何表")
                    
                    if table_name:
                        logger.info(f"开始预览表: {table_name}, 限制: {limit}, 偏移: {offset}")
                        query_result = connection.preview_table_data(table_name, limit, offset)
                        logger.info(f"表预览结果: {query_result}")

                        view_data = {
                            "table_name": table_name,
                            "columns": list(query_result.keys()),
                            "data": pd.DataFrame(query_result.get("data")),
                            "total_records": query_result.get("total_records"),
                            "returned_records": len(query_result.get("data"))
                        }
                        preview_data.update(view_data)
                    else:
                        logger.warning("没有指定表名且无法自动获取表名")
                        
            elif data_source.source_type == "excel":
                # 预览Excel数据
                config = data_source.get_config()
                if config:
                    file_path = config.get("file_path")
                    if file_path:
                        read_result = excel_processor.read_excel_file(file_path)
                        if read_result["success"]:
                            target_sheet = sheet_name or list(read_result["data"].keys())[0]
                            if target_sheet in read_result["data"]:
                                sheet_data = read_result["data"][target_sheet]
                                
                                # 应用分页
                                all_data = sheet_data.get("preview_data", [])
                                total_records = len(all_data)
                                paginated_data = all_data[offset:offset + limit]
                                
                                # 清理数据，将numpy类型转换为Python原生类型
                                clean_columns = self._clean_numpy_types(sheet_data.get("columns", []))
                                clean_data = self._clean_numpy_types(paginated_data)
                                
                                # 确保数据格式符合DataPreviewResponse模型
                                formatted_columns = []
                                for col in clean_columns:
                                    formatted_columns.append({
                                        "name": col.get("name", ""),
                                        "type": col.get("type", "string")
                                    })
                                
                                preview_data.update({
                                    "table_name": target_sheet,
                                    "columns": formatted_columns,
                                    "data": clean_data,
                                    "total_records": total_records,
                                    "returned_records": len(clean_data)
                                })
            
            # 计算执行时间
            execution_time = int((time.time() - start_time) * 1000)
            preview_data["execution_time"] = execution_time
            
            # 记录预览日志
            self._record_preview_log(
                db, 
                data_source_id, 
                "data_preview", 
                "success", 
                preview_data.get("returned_records", 0),
                len(preview_data.get("columns", [])),
                execution_time,
                table_name=table_name,
                sheet_name=sheet_name,
                sample_data=preview_data.get("data", [])[:5]  # 保存前5条记录作为样本
            )
            
            # 更新数据源最后预览时间
            data_source.last_preview_time = datetime.utcnow()
            db.commit()
            
            return preview_data
            
        except Exception as e:
            logger.error(f"预览数据源数据失败: {str(e)}")
            
            # 记录错误日志
            execution_time = int((time.time() - start_time) * 1000) if 'start_time' in locals() else 0
            if 'data_source_id' in locals():
                self._record_preview_log(
                    db, 
                    data_source_id, 
                    "data_preview", 
                    "failed", 
                    0, 
                    execution_time,
                    error_message=str(e),
                    table_name=table_name,
                    sheet_name=sheet_name
                )
            if preview_data:
                return preview_data
            raise DatabaseError(f"预览数据源数据失败: {str(e)}")
    
    def get_preview_history(self, db: Session, data_source_id: int, 
                           page: int = 1, page_size: int = 20) -> Dict[str, Any]:
        """获取预览历史记录"""
        try:
            # 计算分页
            offset = (page - 1) * page_size
            
            # 查询预览历史
            logs_query = db.query(DataSourcePreviewLog).filter(
                DataSourcePreviewLog.data_source_id == data_source_id
            ).order_by(DataSourcePreviewLog.created_at.desc())
            
            total = logs_query.count()
            logs = logs_query.offset(offset).limit(page_size).all()
            
            # 格式化结果
            history = {
                "total": total,
                "page": page,
                "page_size": page_size,
                "total_pages": (total + page_size - 1) // page_size,
                "records": []
            }
            
            for log in logs:
                history["records"].append({
                    "id": log.id,
                    "preview_type": log.preview_type,
                    "table_name": log.table_name,
                    "sheet_name": log.sheet_name,
                    "status": log.status,
                    "record_count": log.record_count,
                    "column_count": log.column_count,
                    "execution_time": log.execution_time,
                    "error_message": log.error_message,
                    "created_at": log.created_at.isoformat()
                })
            
            return history
            
        except Exception as e:
            logger.error(f"获取预览历史记录失败: {str(e)}")
            raise DatabaseError(f"获取预览历史记录失败: {str(e)}")
    
    def update_preview_config(self, db: Session, data_source_id: int, config: Dict[str, Any]) -> DataSource:
        """更新预览配置"""
        try:
            data_source = self.get_data_source(db, data_source_id)
            if not data_source:
                raise ValidationError(f"数据源不存在: {data_source_id}")
            
            # 更新预览配置
            if "preview_enabled" in config:
                data_source.preview_enabled = config["preview_enabled"]
            if "preview_limit" in config:
                data_source.preview_limit = min(config["preview_limit"], 1000)  # 最大限制1000
            if "connection_timeout" in config:
                data_source.connection_timeout = config["connection_timeout"]
            if "max_retry_attempts" in config:
                data_source.max_retry_attempts = config["max_retry_attempts"]
            if "tags" in config:
                data_source.tags = config["tags"]
            if "category" in config:
                data_source.category = config["category"]
            if "priority" in config:
                data_source.priority = config["priority"]
            
            db.commit()
            
            logger.info(f"数据源预览配置更新成功: {data_source.name}")
            return data_source
            
        except Exception as e:
            db.rollback()
            logger.error(f"更新数据源预览配置失败: {str(e)}")
            raise DatabaseError(f"更新数据源预览配置失败: {str(e)}")
    
    def _record_preview_log(self, db: Session, data_source_id: int, preview_type: str,
                          status: str, record_count: int = 0, column_count: int = 0,
                          execution_time: int = 0, error_message: str = None,
                          table_name: Union[str, Dict[str, Any]] = None, sheet_name: str = None,
                          sample_data: List[Dict] = None):
        """记录预览日志"""
        try:
            # 处理表名参数，确保是字符串类型
            processed_table_name = None
            if table_name is not None:
                if isinstance(table_name, dict):
                    processed_table_name = table_name.get('name', str(table_name))
                elif isinstance(table_name, str):
                    processed_table_name = table_name
                else:
                    processed_table_name = str(table_name)
            
            log = DataSourcePreviewLog(
                data_source_id=data_source_id,
                preview_type=preview_type,
                table_name=processed_table_name,
                sheet_name=sheet_name,
                status=status,
                record_count=record_count,
                column_count=column_count,
                execution_time=execution_time,
                error_message=error_message,
                sample_data=sample_data
            )
            
            db.add(log)
            db.commit()
            
        except Exception as e:
            logger.error(f"记录预览日志失败: {str(e)}")
            # 不抛出异常，避免影响主要功能


# 全局数据源管理器
data_source_manager = DataSourceManager()